package com.zhang.flink.exercise;

import com.zhang.flink.bean.ClickEvent;
import com.zhang.flink.example.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;

/**
 * @title: 每隔5s统计过去10s的url访问top3
 * @author: zhangyifan
 * @date: 2022/10/11 10:33
 */
public class UrlCountTopN {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 处理逻辑
        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<ClickEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                                    @Override
                                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .map(r -> r.url)
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10L), Time.seconds(5L)))
                .aggregate(
                        new AggregateFunction<String, HashMap<String, Long>, List<Tuple2<String, Long>>>() {
                            @Override
                            public HashMap<String, Long> createAccumulator() {
                                return new HashMap<>();
                            }

                            @Override
                            public HashMap<String, Long> add(String value, HashMap<String, Long> accumulator) {
                                if (accumulator.containsKey(value)) {
                                    accumulator.put(value, accumulator.get(value) + 1L);
                                } else {
                                    accumulator.put(value, 1L);
                                }
                                return accumulator;
                            }

                            @Override
                            public List<Tuple2<String, Long>> getResult(HashMap<String, Long> accumulator) {
                                ArrayList<Tuple2<String, Long>> list = new ArrayList<>();
                                for (String key : accumulator.keySet()) {
                                    list.add(Tuple2.of(key, accumulator.get(key)));
                                }
                                list.sort(new Comparator<Tuple2<String, Long>>() {
                                    @Override
                                    public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                                        return o2.f1.intValue() - o1.f1.intValue();
                                    }
                                });
                                return list;
                            }

                            @Override
                            public HashMap<String, Long> merge(HashMap<String, Long> a, HashMap<String, Long> b) {
                                for (String key : a.keySet()) {
                                    if (b.containsKey(key)) {
                                        b.put(key, b.get(key) + a.get(key));
                                    } else {
                                        b.put(key, a.get(key));
                                    }
                                }
                                return b;
                            }
                        },
                        new ProcessAllWindowFunction<List<Tuple2<String, Long>>, String, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<List<Tuple2<String, Long>>, String, TimeWindow>.Context context, Iterable<List<Tuple2<String, Long>>> elements, Collector<String> out) throws Exception {
                                List<Tuple2<String, Long>> urlList = elements.iterator().next();
                                StringBuilder result = new StringBuilder("-------------------------" + new Timestamp(context.window().getStart()) + "-------------------------\n");
                                for (int i = 0; i < 3; i++) {
                                    Tuple2<String, Long> tuple2 = urlList.get(i);
                                    result.append("NO."+(i+1)+" url:"+tuple2.f0+",总浏览次数:"+tuple2.f1+"\n");
                                }
                                result.append("-------------------------" + new Timestamp(context.window().getEnd()) + "-------------------------\n");

                                out.collect(result.toString());
                            }
                        }
                )
                .print();


        // 执行任务
        env.execute();

    }
}
